package flink.my_connect;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

public class MySourceConnector {
    public static void main(String[] args) throws Exception{
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> stream = env.addSource(new MySource());
        stream.print();
        env.execute();
    }

    static class MySource implements SourceFunction<String>{
        //用来控制数据产生
        private boolean running = true;

        @Override
        public void run(SourceContext<String> sourceContext) throws Exception {
            Random r = new Random();
            while(running){
                sourceContext.collect(String.valueOf(r.nextGaussian()));//正态分布
                Thread.sleep(1000);//控制输出频率
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}

